/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver.action;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.IdGenerator;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.neptune.tablet.TabletInfo;
import com.nhncorp.neptune.tabletserver.DiskSSTableIF;
import com.nhncorp.neptune.tabletserver.MemorySSTableIF;
import com.nhncorp.neptune.tabletserver.Tablet;
import com.nhncorp.neptune.tabletserver.TabletMapFile;
import com.nhncorp.neptune.tabletserver.TabletServerIF;
import com.nhncorp.neptune.tabletserver.action.TabletSplitAction.TabletSplitFinishAction;

public class MinorCompactionAction extends TabletAction implements MinorCompactionActionIF {
  public static final Log LOG = LogFactory.getLog(MinorCompactionAction.class.getName());
  
  static List<String> checkActions = new ArrayList<String>();
  static {  
    checkActions.add(MinorCompactionActionIF.ACTION_TYPE);
    checkActions.add(MajorCompactionAction.ACTION_TYPE);
    checkActions.add(TabletSplitAction.ACTION_TYPE);
    checkActions.add(TabletSplitFinishAction.ACTION_TYPE);    
    //checkActions.add(TabletStopAction.ACTION_TYPE);
    checkActions.add(ScannerAction.ACTION_TYPE);
    checkActions.add(TabletLockAction.ACTION_TYPE);
    checkActions.add(TabletDropAction.ACTION_TYPE);
  }
  
  protected MinorCompactionActionIF proxyObject;
  protected Tablet tablet;
  protected TabletInfo tabletInfo;
  protected boolean end = false;
  
  public MinorCompactionAction() {
    
  }
  
  public void init(MinorCompactionActionIF proxyObject, NConfiguration conf, Tablet tablet) throws IOException {
    this.proxyObject = proxyObject;
    this.tablet = tablet;
    this.tabletInfo = tablet.getTabletInfo();
  }

  public boolean isWaitingMode() {
    return true;
  }

  public boolean isThreadMode() {
    return true;
  }

  public boolean isEnd() {
    return end;
  }
  
  public void setEnd(boolean end) {
    this.end = end;
  }
  
  public void run() {
    try {
      end = false;
      LOG.info("TabletMinorCompaction Start: " + tabletInfo.getTabletName() + "," + tablet.getTabletServer().getHostName());
      proxyObject.compact();
    } catch (Exception e) {
      LOG.error("run: " + e.getMessage());
    }    
  }
  
  public void compact() throws IOException {
    if(!tablet.isReady() || tablet.getActionChecker().runningActions.contains(TabletSplitAction.ACTION_TYPE)) {
      tablet.getActionChecker().endAction(this);
      LOG.info("TabletMinorCompaction End: " + tablet.getTabletInfo().getTabletName() + " cause by tablet is not ready");
      tablet.setInChangeLogError(false);
      return;
    }
    long startTime = System.currentTimeMillis();
    //minor compaction 처리 중에 memory tablet freeze 시키고 빈 memory tablet 할당할 동안은 commit 처리되면 안됨
    MemorySSTableIF memorySSTable = tablet.getMemorySSTable();
    DiskSSTableIF diskSSTable = tablet.getDiskSSTable();

    Map<String, TabletMapFile> mapFiles = null;
    String fileId = IdGenerator.getId();
    
    try {
      if(memorySSTable.getTabletSize() == 0) {
        return;
      }
      LOG.debug("dataCount:" + memorySSTable.getDataCount()[0] + ", memorySize:" + memorySSTable.getTabletSize());
      
      try {
        //1.MemTable freeze
        memorySSTable.initMemoryForCompaction();
    
        //2.save to temp dir
        //columnName, TabletMapFile      
        mapFiles = tablet.getMemorySSTable().saveToDisk(tabletInfo, fileId);
        
      } catch (Exception e) {
        LOG.error("MinorCompaction Error:" + e.getMessage() + "," + tabletInfo.getTabletName(), e);
  
        tablet.getMemorySSTable().cancelCompaction(fileId);
        
        return;
      } 
      
      //여기서부터 commit이 안되게 한다.
      //MemoryTable lock(move 도중에 read하는 경우 데이터가 중복 조회될 수 있음)
      memorySSTable.obtainWriteLock("compact");
      try {
        //3.move map file from temp dir to table data dir
        for(Map.Entry<String, TabletMapFile> entry: mapFiles.entrySet()) {
          String columnName = entry.getKey();
          TabletMapFile mapFile = entry.getValue();
          proxyObject.moveMapFileFromTemp(diskSSTable, columnName, mapFile);  
        }
      } catch (Exception e) {
        try {
          memorySSTable.releaseWriteLock("compact");
        } catch (Exception err) {
        }
        LOG.fatal("TabletServer shutdowned cause fail to move mapFile from temp after minorCompaction", e);
        tablet.getTabletServer().shutdown();
        return;
      }
      
      
      try {
        memorySSTable.clearCompactingColumnCollections(); 

        //commit lock 해제
        memorySSTable.releaseWriteLock("compact");
        
        //temp dir 삭제
        GPath tempPath = Tablet.getTabletMinorCompactionTempPath(tablet.getConf(), tabletInfo);
        try {
          NeptuneFileSystem fs = NeptuneFileSystem.get(tablet.getConf());
          fs.delete(tempPath, true);
        } catch (IOException e) {
          LOG.error("Can't minor compaction temp dir:" + tempPath, e);
        }
      } catch(Exception e) {
        memorySSTable.releaseWriteLock("compact");
        LOG.fatal("TabletServer shutdowned cause fail to delete ChangeLogFile after minorCompaction:" + e.getMessage(), e);
        tablet.getTabletServer().shutdown(); 
        return;
      }
    } finally {
      // by sangchul
      // If the tablet is in the mode of clearing changelog error, 
      // this two methods are important to do so.
      
      // this method asks chnageLogClient to remove stale changelog location info
      // and make next commitment elect new locations of changelog servers
      memorySSTable.endChangeLogMinorCompaction();
      
      // this method permits to accept next apply operations from clients
      tablet.setInChangeLogError(false);
      
      tablet.getMemorySSTable().setCompacting(false);
      tablet.getActionChecker().endAction(this);
      end = true;
      LOG.info("TabletMinorCompaction End: " + tablet.getTabletInfo().getTabletName() 
          + ", mapFileCount=" + diskSSTable.getMaxMapFileCount() + "," 
          + tablet.getTabletServer().getHostName() 
          + ",time=" + (System.currentTimeMillis() - startTime));
      
      try {
        if(tablet.neededSplit()) {
          tablet.doTabletSplit();
        }
      } catch (IOException e) {
        LOG.error("MajorCompactionAction error while starting split:" + e.getMessage(), e);
      }
    }
  }
  
  public void moveMapFileFromTemp(DiskSSTableIF diskSSTable, String columnName, TabletMapFile mapFile ) throws IOException {
    GPath targetPath = new GPath(Tablet.getTabletPath(tablet.getConf(), tablet.getTabletInfo()), columnName + "/" + mapFile.getFileId() + "/");
    mapFile.moveTo(targetPath, true);
    diskSSTable.addTabletMapFile(columnName, mapFile);
  }
  
  public String getActionKey() {
    return tablet == null ? "null:" + getActionType() : tablet.getTabletInfo().getTabletName() + getActionType();

    //return getActionType();
  }

  public List<String> getRunCheckActionTypes() {
    return checkActions;
  }
  
  public List<String> getWaitCheckActionTypes() {
    List<String> actions = new ArrayList<String>();
    actions.add(TabletSplitFinishAction.ACTION_TYPE);
    actions.add(TabletSplitAction.ACTION_TYPE);
    return actions;
  }

  @Override
  public String getActionType() {
    return ACTION_TYPE;
  }

  public String getTestHandlerKey() {
    return tablet.getTabletServer().getHostName();
  }
  
  public TabletServerIF getTabletServer() {
    return tablet.getTabletServer();
  }
}
